package com.atguigu.flink.chapter11.function;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/10/23 9:09
 */
public class Flink02_Function_Table {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> stream =
            env.fromElements("hello hello", "hello world", "atguigu hello hello");
        
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        Table table = tenv.fromDataStream(stream);
        table.printSchema();
        
        // 1. 在table api中使用
        
        // 1.1 内联的方式
        /*table
            .joinLateral(call(MySplit.class, $("f0")))
            .select($("f0"), $("word"), $("len"))
            .execute()
            .print();*/
        // 1.2 注册后使用
        //        tenv.createTemporaryFunction("my_split", MySplit.class);
        
       /* table
            .leftOuterJoinLateral(call("my_split", $("f0")))
            //            .joinLateral(call("my_split", $("f0")))
            .select($("f0"), $("word"), $("len"))
            .execute()
            .print();*/
        // 2. 在sql语句中使用
        // 2. 1先注册
        tenv.createTemporaryFunction("my_split", MySplit.class);
        // 2.2 再使用
        /*tenv.sqlQuery("select" +
                          " f0, " +
                          " word, " +
                          " len " +
                          "from " + table +
                          " join lateral table(my_split(f0)) on true"
        ).execute().print();*/
        // select .. from a join b on a.id=b.id
        // select .. from a,b where a.id=b.id  内连接的第二种写法
        tenv.sqlQuery("select" +
                          " f0, " +
                          " word, " +
                          " len " +
                          "from " + table +
                          "  ,lateral table(my_split(f0))"
        ).execute().print();
    
       /* tenv.sqlQuery("select" +
                          " f0, " +
                          " word, " +
                          " len " +
                          "from " + table +
                          " left join lateral table(my_split(f0)) on true"
        ).execute().print();*/
    
        /*tenv.sqlQuery("select" +
                          " f0, " +
                          " w, " +
                          " l " +
                          "from " + table +
                          " left join lateral table(my_split(f0)) as T(w, l) on true"
        ).execute().print();*/
        
    }
    
    @FunctionHint(output = @DataTypeHint("row<word string, len int>"))
    public static class MySplit extends TableFunction<Row> {
        public void eval(String s) {
            
            if (s.contains("atguigu")) {
                return;
            }
            
            String[] words = s.split(" ");
            
            for (String word : words) {
                collect(Row.of(word, word.length()));
            }
        }
    }
    
}
/*
hello hello                hello     5
                           hello     5
           
hello world                hello      5
                           world      5
                        
atguigu hello hello       atguigu     7
                            ....


 */